/**
 * GeneralServiceMessageDispatcher.java
 * ************************** 
 * @date Apr 11, 2011
 * @author Michele Tamburini
 * 
 * **************************
 */
package gossipServices.network;

import gossipServices.gossipMessages.GeneralGossipMessage;

import java.util.ArrayDeque;
import java.util.Queue;

import networkRmi.NetworkManager;

import messages.Message;

/**
 * Description:
 * The GeneralServicesMessageDispatcher is a standard dispatcher
 * for the incoming messages through the NetworkManager.
 * It represent a standard dispatcher for a target service
 * that relies on it.
 *  
 * In order to build an instance of this class the caller
 * has to provide three Class<?> parameters
 * in the constructor that represent:
 *   1) "messageClass":
 *   	the class of the Message(s) directed to the target
 *   	service,
 *   
 *   2) "toActiveThreadClass":
 *   	the class that will send Message(s) to the active thread
 *   	(generally the passive thread in an external node)
 *    
 *   3) "toPassiveThreadClass":
 *   	the class that will send Message(s) to the passive thread
 *   	(generally the active thread in an external node)
 *    
 * It implements the MessageHandler interface. 
 * The extractServiceMessages(Queue<Message> queue) function
 * extract first all messages directed to the target service,
 * comparing the incoming list with "messageClass".
 * Then it separates the Message(s) that have to be forwarded
 * to the active thread by comparing them with "toActiveThreadClass",
 * from the ones directed to the passive thread "toPassiveThreadClass".
 * 
 *  The AggregationNetworkManager is an example of the usage
 *  of this class.
 *
 */
public class GeneralServicesMessageDispatcher extends GeneralServicesNetManager {
	
	protected Class<?> messageClass;
	protected Class<?> toActiveThreadClass;
	protected Class<?> toPassiveThreadClass;
	
	protected Queue<Message> rawReceivedMessages = new ArrayDeque<Message>(200);

	/**
	 * @param networkManager
	 */
	public GeneralServicesMessageDispatcher(NetworkManager networkManager, Class<?> messageClass, 
			Class<?> toActiveThreadClass, Class<?> toPassiveThreadClass) {
		super(networkManager);
		this.messageClass = messageClass;
		this.toActiveThreadClass = toActiveThreadClass;
		this.toPassiveThreadClass = toPassiveThreadClass;		
	}

	@Override
	public void extractIncomingMessages(Queue<Message> queue) {
		int size = queue.size();
		int updatedRawSize;
		
		for(int i=0; i<size;i++){
			Message m = queue.poll();
			if(messageExtractionAlgorithm(m) ){
				rawReceivedMessages.add(m);
			}else{
				queue.add(m);
			}			
		}
						
		/*
		 * Warning: two thread are running on the two
		 * vectors of active and passive received messages  
		 */
		updatedRawSize = rawReceivedMessages.size();
		for(int i=0; i<updatedRawSize; i++){
			GeneralGossipMessage msg = (GeneralGossipMessage) rawReceivedMessages.poll();
			if(msg.getGossipThread().equals(toActiveThreadClass)){
				/*
				 * Inserting in the Active Thread Messages
				 */
				synchronized (activeReceivedMessages) {					
					activeReceivedMessages.add(msg);
					activeReceivedMessages.notify();
				}
			}else if(msg.getGossipThread().equals(toPassiveThreadClass)){
				/*
				 * Inserting the Passive Thread Messages
				 */
				synchronized (passiveReceivedMessages) {
					passiveReceivedMessages.add(msg);
					passiveReceivedMessages.notify();
				}
			}else{
				System.err.println(this.getClass()+" unable to dispatch "+msg);
				rawReceivedMessages.add(msg);
			}
		}
		
//		System.out.println(this.getClass().getSimpleName()+" messages Queues:\n"+
//				"active: "+activeReceivedMessages.size()+"\n"+
//				"waiting: "+waitingIncoming.size()+"\n"+
//				"passive: "+passiveReceivedMessages.size());
		
		
	}

	/**
	 * @param m
	 * @return
	 */
	protected boolean messageExtractionAlgorithm(Message m) {
		return  m.getClass().equals(messageClass);
	}

}
